package com.xzkj.rx.thread.utils;

import android.annotation.SuppressLint;
import android.os.Handler;
import android.os.HandlerThread;
import android.os.Looper;

import androidx.annotation.NonNull;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

/**
 * Description: rxjava2的简化封装
 * Data：2021/12/8-上午11:09
 * @author liujie
 */
@SuppressLint("CheckResult")
public class Rx {

    /**
     * 获取main线程的调度器
     * @return
     */
    public static Scheduler main(){
        return AndroidSchedulers.mainThread();
    }

    /**
     * 获取io调度器
     * @return
     */
    public static Scheduler io(){
        return Schedulers.io();
    }

    /**
     * 简化创建Observable，常用场景，子线程中执行耗时任务，完成后自动切换到主线程
     * @param source
     * @param <T>
     * @return
     */
    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
        return Observable.create(source)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    /**
     * 简化创建集合遍历的Observable
     * @param iterable
     * @param <T>
     * @return
     */
    public static <T> Observable<? extends T> create(Iterable<? extends T> iterable) {
        return Observable.fromIterable(iterable)
                .subscribeOn(Schedulers.io());
    }

    /**
//     * 注册的handler主线程监听
//     */
//    public static ConcurrentHashMap<String,HandleCall> handlerCalls = new ConcurrentHashMap<>();
//    /***
//     * 注册的handler的key，通过key来找到需要回调的接口
//     */
//    public static ConcurrentHashMap<String,String> handlerKeys = new ConcurrentHashMap<>();

    /**
     * 主线程handler
     */
    public static final Handler MAIN_HANDLER = new Handler(Looper.getMainLooper());

    /**
     * 在子线程中执行耗时任务, 把运算结果返回给主线程，回调到主线程
     *
     * @param rxCallBack
     * @param <T>
     */
    public static <T> void runSub(final RxCall<T> rxCallBack) {
        Observable.create((ObservableOnSubscribe<T>) emitter -> {
            //onNext中的参数不能为null，否则onNext接收不到
            emitter.onNext(rxCallBack.doSub());
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<T>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull T t) {
                rxCallBack.onUi(t);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    /**
     * 一个子线程顺序执行任务，把每次执行的结果抛给主线程
     * @param rxCallBack
     * @param <T>
     */
    public static <T> void runSubFlow(final RxMultiCall<T> rxCallBack) {
        Observable.create((ObservableOnSubscribe<T>) emitter -> {
            rxCallBack.doSub(emitter);
            emitter.onComplete();
        }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Observer<T>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull T t) {
                rxCallBack.onUi(t);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                rxCallBack.onError(e);
            }

            @Override
            public void onComplete() {
                rxCallBack.onComplete();
            }
        });
    }

    /**
     * 切换线程，切换到子线程
     * @param runnable
     */
    public static void runSub(Runnable runnable) {
        Observable.create(emitter -> {
            runnable.run();
        }).subscribeOn(Schedulers.io()).subscribe();
    }

    /**
     * 单个线程池调度
     * @param runnable
     */
    public static void runSingle(Runnable runnable) {
        Observable.create(emitter -> {
            runnable.run();
        }).subscribeOn(Schedulers.single()).subscribe();
    }

    /**
     * 切换到子线程，在子线程中延迟执行任务
     * @param runnable
     * @param delayTime
     */
    public static void runSubDelay(Runnable runnable, long delayTime) {
        Observable.create(emitter -> {
            try {
                Thread.sleep(delayTime);
                runnable.run();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).subscribeOn(Schedulers.io()).subscribe();
    }

    /**
     * 切换到主线程，在主线程中延迟执行任务
     * @param runnable
     * @param delayTime
     */
    public static void runUiDelay(Runnable runnable, long delayTime) {
        MAIN_HANDLER.postDelayed(runnable, delayTime);
    }

    /**
     * 切换到主线程
     * @param runnable
     */
    public static void runUi(Runnable runnable) {
        if (Looper.myLooper() == Looper.getMainLooper()){
            runnable.run();
        }else{
            MAIN_HANDLER.post(runnable);
        }
    }

    /**
     * 取消延迟执行的任务
     * @param runnable
     */
    public static void removeCallbacks(Runnable runnable) {
        MAIN_HANDLER.removeCallbacks(runnable);
    }

//    public static void setMsgCall(String key,HandleCall handlerCall){
//        handlerCalls.put(key,handlerCall);
//    }
//
//    public static void sendMsg(String key,Message msg){
//        handlerKeys.put(key,"");
//        MAIN_HANDLER.sendMessage(msg);
//    }
//
//    public static void removeMsgCall(String key){
//        handlerKeys.remove(key);
//        handlerCalls.remove(key);
//    }

    /**
     * 每隔interval毫秒，向主线程发送一个消息
     * @param interval
     * @param rxInterval
     */
    public static void interval(long interval, RxInterval rxInterval){
        Observable.interval(interval, TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable disposable) {
                        rxInterval.setDisposable(disposable);
                    }

                    @Override
                    public void onNext(@NonNull Long number) {
                        rxInterval.onUi(number,rxInterval.disposable);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    /**
     * handler子线程
     */
    public static HandlerThread rxHandlerThread = new HandlerThread("rx_handler");
    /**
     * 子线程中运行的handler
     */
    public static Handler rxHandler;

    /**
     * 默认开启子线程循环队列
     */
    static {
        rxHandlerThread.start();
        rxHandler = new Handler(rxHandlerThread.getLooper());
    }

    /**
     * 后台处理任务，该函数使用单个线程开启loop队列。一般用于处理毫秒级任务，用于需要立即处理的任务。处理完毕之后立即刷新UI的任务。
     * 超过1-5秒以上的任务使用线程池
     * @see #runSub(Runnable runnable)
     * @param runnable
     */
    public static void loopMsg(Runnable runnable){
        if (null == rxHandler){
            rxHandler = new Handler(rxHandlerThread.getLooper());
        }
        rxHandler.post(runnable);
    }

    /**
     * 子线程中执行延迟消息，真实执行的后台任务建议不要超过1-5s。
     * @param runnable
     * @param delayTime 毫秒
     */
    public static void loopMsgDelay(Runnable runnable,long delayTime){
        if (null == rxHandler){
            rxHandler = new Handler(rxHandlerThread.getLooper());
        }
        rxHandler.postDelayed(runnable,delayTime);
    }
}
